/*
* Copyright (C) 2010-2012 Magnus Skjegstad
* 
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version 2
* of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
* 
* The latest version of this library can be downloaded from http://mist-pubsub.googlecode.com
* 
*/
package com.skjegstad.mist2.modules.udp.broadcast;

import com.skjegstad.simpleevents.SimpleAsyncEventListener;
import com.skjegstad.simpleevents.SimpleEvent;
import com.skjegstad.simpleevents.SimpleEventListener;
import com.skjegstad.simpleevents.SimpleEventPipe;
import com.skjegstad.mist2.IMistNetworkModule;
import com.skjegstad.mist2.MistUtil;
import com.skjegstad.mist2.NetworkModuleHelper;
import com.skjegstad.mist2.datatypes.Acknowledgement;
import com.skjegstad.mist2.datatypes.BloomfilterAcknowledgment;
import com.skjegstad.mist2.datatypes.ConnectionIdentifier;
import com.skjegstad.mist2.datatypes.DataFragment;
import com.skjegstad.mist2.datatypes.ListAcknowledgement;
import com.skjegstad.mist2.datatypes.NodeIdentifier;
import com.skjegstad.mist2.datatypes.TopicIdentifier;
import com.skjegstad.mist2.datatypes.TopicMap;
import com.skjegstad.mist2.datatypes.TopicMapEntry;
import com.skjegstad.mist2.events.NewDataElementReceivedEvent;
import com.skjegstad.mist2.events.NewDataFragmentsReceived;
import com.skjegstad.mist2.events.NewNodeDiscoveredEvent;
import com.skjegstad.mist2.events.NewSubscriptionEvent;
import com.skjegstad.mist2.events.RegisterNodeEvent;
import com.skjegstad.mist2.events.RemoveNodeEvent;
import com.skjegstad.mist2.exceptions.MistException;
import com.skjegstad.mist2.managers.MistEventManager;
import com.skjegstad.mist2.modules.udp.BeaconMessage;
import com.skjegstad.mist2.modules.udp.DataFragmentMessage;
import com.skjegstad.mist2.modules.udp.MessageContainer;
import java.io.Externalizable;
import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import no.ffi.xudp.interfaces.XUdpThreadExceptionCallback;
import no.ffi.xudp.interfaces.XUdpThreadStoppedCallback;
import no.ffi.xudp.threads.XUdpMulticastListener;
import no.ffi.xudp.threads.XUdpMulticastSender;
import no.ffi.xudp.threads.XUdpUnicastListener;
import no.ffi.xudp.threads.XUdpUnicastSender;
import no.ffi.xudp.threads.common.XUdpGenericListener;
import no.ffi.xudp.threads.common.XUdpGenericSender;

/**
 *
 * @author Magnus Skjegstad
 */
public class UDPBroadcastModule implements IMistNetworkModule, XUdpThreadExceptionCallback, XUdpThreadStoppedCallback {
    private NetworkModuleHelper networkModuleHelper;
    private int fragmentSize;
    private ConnectionIdentifier connectionIdentifier;
    private NodeIdentifier nodeIdentifier;
    private Random r = new Random();
    private ExecutorService executor;
    private BlockingQueue<DatagramPacket> receiveQueue;
    private BlockingQueue<DatagramPacket> sendQueue;
    private MistEventManager eventManager;
    private SimpleEventPipe localEventPipe;
    private long lastBeaconSent = 0;
    private long hashesInLastAck = -1; // update extrabits when the number of hashes is the same in two bloom filters
    private byte extraBits = 13;
    private int k = 1;

    private boolean optimizeForShortMessages = true; // When this is true, ack's are sent less frequently at the expense of more frequent retransmissions. This makes sense when the ack is larger than the message.

    private int periodicBeaconTime = 60000; // every minute
    private int timerTick = 1000;
    private boolean shouldSendBeacon = true;
    private long rate = 1000; // hashes to send per tick
    private long minimumRate = 1350; // ca 1 mtu
    private int maxAvailableHashes = 100; // max hashes that have been available at any moment
    private int maxHashesToSendAtOnce = 1000; // This could be set dynamically in the future
    private int lastSent = -1; // messages sent during the last call to sendDataFragments
    private boolean enableRateControl = false; // Set to true to enable automatic transmission rate control
    private boolean randomizeFragments = true; // Whether to randomize fragments before sending them. This may improve cooperative sync.
    private boolean useListAcknowledgments = false; // Whether to disable bloom filters and just use lists of hashes to acknowledge

    private final int listenPort;
    private final InetAddress sendAddress;

    private final Map<NodeIdentifier, Long> lastSentFragments = new HashMap();
    private final Map<NodeIdentifier, Long> nodeRetries = new HashMap();
    private final Map<NodeIdentifier, Integer> lastMissing = new HashMap();
    private int maxNodeRetries = 3;
    private int minRetryDelay = timerTick * 4;

    private XUdpGenericListener listener;
    private XUdpGenericSender sender;

    private boolean useMulticast = false;

    private NetworkInterface networkInterface;

    /**
     * Initialize UDP broadcast module.
     *
     * @param address address to send multicast or broadcast messages on.
     * @param listenPort port to use for broadcast messages.
     * @param useMulticast when true, use multicast (MTU 64000), when false, use broadcast (MTU 1400)
     * @param networkInterface network interface to use for multicast. May be null.
     */
    public UDPBroadcastModule(InetAddress address,
            int listenPort,
            boolean useMulticast,
            Properties config) throws MistException {                
        this.sendAddress = address;
        this.listenPort = listenPort;

        this.useMulticast = useMulticast;

        if (useMulticast)
            this.fragmentSize = 64000;
        else
            this.fragmentSize = 1100;
        
        if (config != null) {
            // Set maximum fragment size. When using transports without IP layer fragmentation, this should be set 
            // lower than the MTU. 
            if (config.getProperty("fragmentSize") != null) {
                try {
                    this.fragmentSize = Integer.parseInt(config.getProperty("fragmentSize"));
                } catch (NumberFormatException ex) {
                    throw new MistException("Invalid fragment size", ex);
                }
            }
            
            // Whether to use a List of hashes as acknowledgment instead of a Bloom filter. Intended for bandwidth comparison.
            if (config.getProperty("useListAcknowledgments") != null) {
                try {
                    this.useListAcknowledgments = Boolean.parseBoolean(config.getProperty("useListAcknowledgments"));
                    if (this.useListAcknowledgments)
                        System.err.println("WARNING: Using Lists of hashes as acknowledgments (not minimal Bloom filters)");
                } catch (Exception ex) {
                    throw new MistException("useListAcknowledgments must be true or false",ex);
                }
            }

            // Max number of fragments/data elements to send per data exchange
	    if (config.getProperty("maxElementsPerTick") != null) {
	       try {
		    this.maxHashesToSendAtOnce = Integer.parseInt(config.getProperty("maxElementsPerTick"));
	       } catch (NumberFormatException ex) {
		   throw new MistException("Invalid value for maxElementsPerTick", ex);
	       }
	    }
            
            // When this is true, ack's are sent less frequently at the expense of more frequent retransmissions. This makes sense when the ack is larger than the message.
            if (config.getProperty("optimizeForShortMessages") != null) {
                try {
                    this.optimizeForShortMessages = Boolean.parseBoolean(config.getProperty("optimizeForShortMessages"));
                } catch (Exception ex) {
                    throw new MistException("Invalid format: optimizeForShortMessages", ex);
                }
            }

            // Use experimental rate control to avoid network congestion (may not always work as expected)
            if (config.getProperty("rateControl") != null) {
                try {
                    this.enableRateControl = Boolean.parseBoolean(config.getProperty("rateControl"));
                } catch (Exception ex) {
                    throw new MistException("Invalid value for rate control (true/false)", ex);
                }
            }
            
            // Whether to randomize the order in which messages are sent. This is to reduce the probability
            // of two neighbors sending the same data element at the same time.
            if (config.getProperty("randomizeFragments") != null) {
                try {
                    this.randomizeFragments = Boolean.parseBoolean(config.getProperty("randomizeFragments"));
                } catch (NumberFormatException ex) {
                    throw new MistException("Invalid value for randomizeFragments (true/false)", ex);
                }
            }
            if (config.getProperty("networkInterfaceName") != null) {
                try {
                    this.networkInterface = NetworkInterface.getByName(config.getProperty("networkInterfaceName"));
                } catch (SocketException ex) {
                    throw new MistException("Unknown network interface", ex);
                }
            }

            if (config.getProperty("networkInterfaceIp") != null) {
                try {
                    this.networkInterface = NetworkInterface.getByInetAddress(InetAddress.getByName(config.getProperty("networkInterfaceIp")));
                } catch (Exception ex) {
                    throw new MistException("Unknown network interface", ex);
                }
            }

            // When to send periodic beacon after the last message has been received.
            if (config.getProperty("periodicBeaconTime") != null) {
                try {
                    periodicBeaconTime = Integer.parseInt(config.getProperty("periodicBeaconTime"));
                } catch (Exception ex) {
                    throw new MistException("Unable to parse config variable periodicBeaconTime", ex);
                }
            }
        }

    }

    public UDPBroadcastModule(InetAddress address, int listenPort, boolean useMulticast) throws MistException {
        this(address, listenPort, useMulticast, null);
    }

    protected void setupUDPUnicastThreads() {
        receiveQueue = new LinkedBlockingQueue<DatagramPacket>();
        sendQueue = new LinkedBlockingQueue<DatagramPacket>();

        listener = new XUdpUnicastListener(receiveQueue, null, listenPort, null);
        sender = new XUdpUnicastSender(sendQueue, 0, null);
    }

    protected void setupUDPMulticastThreads() {
        receiveQueue = new LinkedBlockingQueue<DatagramPacket>();
        sendQueue = new LinkedBlockingQueue<DatagramPacket>();

        listener = new XUdpMulticastListener(sendAddress, listenPort, networkInterface, receiveQueue, null);
        sender = new XUdpMulticastSender(sendAddress, listenPort, 1, networkInterface, sendQueue, 0, null);
    }

    public void start(final NetworkModuleHelper networkModuleHelper) throws MistException {
        executor = Executors.newCachedThreadPool();
        eventManager = networkModuleHelper.getMistEventManager();
        localEventPipe = eventManager.getUniquePipe("UDP-BCAST-MODULE");
        
        this.networkModuleHelper = networkModuleHelper;
        
        extraBits = -1; // ugly hack
        while (extraBits <= 0)        
            extraBits = (byte) (r.nextInt() % networkModuleHelper.getMaximumExtraBits()); // set extra bits randomly to increase chance of delivery with multiple neighbors        
        networkModuleHelper.debug("Initial extra bits is " + extraBits);

        if (useMulticast)
            setupUDPMulticastThreads();
        else
            setupUDPUnicastThreads();        
        
        resetRate();
        
        listener.registerThreadExceptionCallback(this);
        listener.registerThreadStoppedCallback(this);
        
        executor.submit(listener);

        sender.registerExceptionCallback(this);
        sender.registerThreadStoppedCallback(this);

        executor.submit(sender);       
        
        executor.submit(new Runnable() {
            public void run() {
                try {
                    while (!Thread.interrupted()) {
                        DatagramPacket p = receiveQueue.take();            
                        receiveUDPPacket(p);
                    }
                } catch (InterruptedException ex) {
                    eventManager.throwException(new MistException("Interrupted while listening for packets", ex));
                    return;
                } catch (Exception ex) {
                    eventManager.throwException(new MistException("Unexpected exception while listening for udp packets", ex));
                    return;
                }
            }
        });

        registerEvents();
        
        return;
    }

    public void stop() {
        executor.shutdownNow();
        // TODO should deregister event listeners...
    }

    private int randomMs(int ms, double perm) {
        double part = ms * perm;
        int ret = (int)Math.round(part + (r.nextDouble() * ms));
        //System.err.println(ret);
        return ret;
    }

    private DatagramPacket classToDatagram(Externalizable c) {
        try {
            return MistUtil.classToDatagram(c, sendAddress, listenPort);
        } catch (Exception ex) {
            eventManager.throwException(new MistException("Unable to serialize class", ex));
        }
        return null;
    }
    
    private void sendMessage(Externalizable message) throws ClassNotFoundException {
        sendQueue.add(classToDatagram(new MessageContainer(message)));
    }

    private void sendBeacon() throws MistException {
        try {
            Set<TopicIdentifier> subscribedTopics = networkModuleHelper.getActiveTopics().keySet();
            /*if (subscribedTopics == null || subscribedTopics.isEmpty()) {
                networkModuleHelper.debug("No subscribers. Beacon not sent.");
                return;
            }*/
                        
            MistUtil.debug("Active topics:");
            MistUtil.debug("--------------");
            for (Entry<TopicIdentifier, TopicMapEntry> e : networkModuleHelper.getActiveTopics().entrySet()) {                
                MistUtil.debug(e.getKey().toString() + ", ttl: " + e.getValue().getTtl() + " checksum:" + e.getValue().getChecksum());
            }
                
            updateK(); // update k based on number of neighbors
                                    
            TopicMap topics = networkModuleHelper.getActiveTopics();
            Collection<Long> hashes = networkModuleHelper.getHashesToAck(subscribedTopics, false, false); // include expired and invalid items in beacon

            if (hashes.size() == hashesInLastAck && !hashes.isEmpty()) {
                extraBits = (byte) ((extraBits + k) % networkModuleHelper.getMaximumExtraBits());
                if (extraBits < 0)
                    throw new MistException("Error: extraBits < 0");
            }  else
                hashesInLastAck = hashes.size();

            if (hashes.isEmpty())
                networkModuleHelper.debug("Warning: Sending empty bloom filter");

            BeaconMessage bm;
            int msg_len;
            do {
                Acknowledgement ack;
                if (!useListAcknowledgments)
                    ack = new BloomfilterAcknowledgment(hashes, extraBits, k);
                else {
                    networkModuleHelper.debug("ListAcknowledgment created.");
                    ack = new ListAcknowledgement(hashes);
                }
                bm = new BeaconMessage(networkModuleHelper.getMyNodeIdentifier(),
                    topics, networkModuleHelper.getMaximumAge(),
                    -1, // everybody use the same port
                    ack);
                msg_len = classToDatagram(new MessageContainer(bm)).getLength();
                if (msg_len > fragmentSize) {
                    networkModuleHelper.debug("Beacon to large. Reducing k.");
                    if (k > 1)
                        k--;
                }
            } while (msg_len > fragmentSize && k > 1);

            lastBeaconSent = System.currentTimeMillis();

            networkModuleHelper.debug("Sending beacon w/k=" + k + " len:" +msg_len);

            sendMessage(bm);
            
        } catch (Exception ex) {
            throw new MistException("Unable to send beacon", ex);
        }
    }
    
    private void resetRate() {
        if (!enableRateControl)
            return;
        rate = fragmentSize;
        sender.setBytesPerSecond(rate);
        MistUtil.debug("Rate reset to " + rate);
    }
    
    private void increaseRate() {
        if (!enableRateControl)
            return;
        rate = rate + (fragmentSize/8);
        sender.setBytesPerSecond(rate);
        MistUtil.debug("Rate increased to " + rate);
    }
    
    private void decreaseRate() {
        if (!enableRateControl)
            return;
        rate = rate / 2;
        if (rate < minimumRate) {
            rate = minimumRate;
        }
        sender.setBytesPerSecond(rate);
        MistUtil.debug("Rate decreased to " + rate);
    }

    private void sendDataFragments(boolean sendBeaconWithFragments) throws MistException {
        try {                                               
            
            // Discard old elements from queue;
            int del =  sender.getQueueLength();
            sender.clearQueue(); // remove unsent stuff
            int deltaSent = lastSent - del; // number of messages sent this round

            if (del > 0)
                networkModuleHelper.debug("Discarded approx. " + del + " items from sendQueue");
            
            // Should we send beacon? This is added here, so that we don't end up deleting beacons from the queue all the time.
            // Beacons should not always be sent with fragments, e.g. when faster acknowledgments are required.
            if (sendBeaconWithFragments) {
                if (shouldSendBeacon) {
                    shouldSendBeacon = false;
                    try {
                        sendBeacon();
                        networkModuleHelper.debug("Ack beacon sent.");
                    } catch (Exception ex) {
                        eventManager.throwException(new MistException("Unable to send beacon.", ex));
                    }
                }
           }
            
            // Find new elements to send
            Set<Long> c = new HashSet();
            // synchronize iterator, just in case            
            Collection<NodeIdentifier> neighbors = networkModuleHelper.getNodes();
            synchronized(neighbors) {
                
                boolean currentRateIsReceivedByAll = true; // Are all neighbors able to receive current rate?
                
                for (NodeIdentifier node : neighbors) {
                    // remove node if it has too many retries
                    if (nodeRetries.containsKey(node) && nodeRetries.get(node) >= maxNodeRetries) {
                        networkModuleHelper.debug(("Node " + node + " has not responded for " + maxNodeRetries + ". Removing as neighbor."));
                        networkModuleHelper.removeNeighborNode(node);
                        nodeRetries.remove(node);
                        lastSentFragments.remove(node);
                        lastMissing.remove(node);
                        decreaseRate();
                        continue;
                    }
                    // don't send if it is less than minRetryDelay milliseconds since last time for this node
                    if (lastSentFragments.containsKey(node)) {
                        if (System.currentTimeMillis() - lastSentFragments.get(node) < minRetryDelay) {
                            //networkModuleHelper.debug("Waiting to send data to node " + node + ". Ack not received yet. Rate=" + rate);
                            continue;
                        } else {
                            	currentRateIsReceivedByAll = false;
				networkModuleHelper.debug("Rate reduced, caused by node " + node.getNodeIdentifier());
                        }
                    }

                    // Add the hashes
                    Collection<Long> hashes = networkModuleHelper.getHashesMissingInNode(node, true, true); // exclude invalid and expired 
                    
                    // Check if all we sent was received
                    int deltaReceived = 0;
                    if (lastMissing.get(node) != null)
                        deltaReceived = lastMissing.get(node) - hashes.size();
                    if (deltaReceived < deltaSent && deltaReceived > 0 && deltaSent > 0) {
                        currentRateIsReceivedByAll = false;
                    }
                    lastMissing.put(node, hashes.size()); // Update with new value                    

                    if (hashes != null && !hashes.isEmpty()) {
                        c.addAll(hashes);
                        lastSentFragments.put(node, System.currentTimeMillis());
                        if (nodeRetries.get(node) != null) {
                            nodeRetries.put(node, nodeRetries.get(node)+1);
                        } else
                            nodeRetries.put(node, new Long(1));
                    }
                }
                
                // Adjust rate if some messages in last round were not received
                if (neighbors != null && neighbors.size() > 0 && deltaSent > 0)
                    if (!currentRateIsReceivedByAll) {
                        decreaseRate();
                    } else {
                        increaseRate();
		    }
            }
            
            
            if (c.isEmpty()) {
                lastSent = 0;
                return;
            }

            List<Long> hashes = new ArrayList(c);
            if (c.size() > maxAvailableHashes)
                maxAvailableHashes = c.size();
            
            Collections.shuffle(hashes); // TODO would it reduce jitter if we chose with lowest seq nr first?

            // remove after shuffle
	    if (maxHashesToSendAtOnce > 0)
	        while (hashes.size() > maxHashesToSendAtOnce) // reduce hashes we send to the target rate
        	      hashes.remove(0);
            
            List<DataFragment> fragments = (List<DataFragment>) networkModuleHelper.getDataFragments(hashes, true, true); // exclude invalid and expired
            lastSent = fragments.size();
                        
            if (fragments.isEmpty()) {
                return;
            }
            
            // Sort fragments by sequence number to reduce jitter
            if (!randomizeFragments) {
                Collections.sort((List)fragments, new Comparator<DataFragment>() {
                    public int compare(DataFragment o1, DataFragment o2) {
                        if (o1.getSeqNr() > o2.getSeqNr())
                            return 1;
                        if (o2.getSeqNr() < o2.getSeqNr())
                            return -1;
                        return 0;
                    }
                });
                if (fragments.size() > 1)
                    if (fragments.get(0).getSeqNr() > fragments.get(1).getSeqNr()) // Fragments must be ordered the correct way
                        throw new MistException("fragments ordered the wrong way");           
            }
            
            networkModuleHelper.debug("About to send " + fragments.size() + " data fragments");            
            
            for (DataFragment f : fragments) {
                try {
                    DataFragmentMessage m = new DataFragmentMessage(f);
                    //networkModuleHelper.debug("Fragment: " + new String(m.getFragment().getPayload()));
                    sendMessage(m);
                } catch (Exception ex) {
                    throw new MistException("Unable to send data fragment " + f.toString(), ex);
                }
            }
            
        } catch (Exception ex) {
            throw new MistException("Unable to send data fragments", ex);
        }
    }

    private void updateK() {
        /* int nodes = networkModuleHelper.getNodes().size(); // +1 is us
        if (nodes > 0) {
            int j = (int) Math.ceil(8.0 / nodes);
            if (j < 1)
                k = 1;
            else
                k = j;
        } else
            k = 8; */
		k = 8;
    }
    
    private void receiveUDPPacket(DatagramPacket packet) {
        try {                    
            MessageContainer mc = new MessageContainer();
            MistUtil.datagramToClass(packet.getData(), mc);

            if (mc.getMessageObject() instanceof BeaconMessage) {
                receiveBeaconMessage((BeaconMessage)mc.getMessageObject(), packet.getAddress());
            } else
            if (mc.getMessageObject() instanceof DataFragmentMessage) {
                DataFragmentMessage m = (DataFragmentMessage) mc.getMessageObject();

                // skip our own fragments
                if (m.getFragment().getSourceNode().equals(networkModuleHelper.getMyNodeIdentifier()))
                    return;

                networkModuleHelper.addDataFragments(Collections.singletonList(m.getFragment()));
                //networkModuleHelper.debug("New data fragment received. Sending beacon.");
                //shouldSendBeacon = true;
            } else
                eventManager.throwException(new MistException("Unknown message type received!"));

        } catch (Exception ex) {
            eventManager.throwException(new MistException("Unable to deserialize datagram packet", ex));
            return;
        }
    }
    
    private void receiveBeaconMessage(BeaconMessage m, InetAddress sourceAddress) {
        try {
            if (!m.getNodeIdentifier().equals(networkModuleHelper.getMyNodeIdentifier())) {               
                networkModuleHelper.debug("Beacon received from " + m.getNodeIdentifier() + ", bits:" + m.getAcknowledgements().getLengthInBits());

                // reset retry counter
                nodeRetries.put(m.getNodeIdentifier(), new Long(0));
                lastSentFragments.remove(m.getNodeIdentifier());

                // register last seen
                eventManager.getNodeManagerEventPipe().trigger(new RegisterNodeEvent(connectionIdentifier,
                    m.getNodeIdentifier(),
                    sourceAddress,
                    m.getSubscribeToTopics(),
                    m.getAcknowledgements()));
            }
        } catch (Exception ex) {
            eventManager.throwException(new MistException("Unexpected exception while receiving beacon", ex));
        }
    }

    private void registerEvents() {

        // On new data element registered in data store
        eventManager.getDataManagerEventPipe().listen(new SimpleEventListener() {
            public void receiveEvent(SimpleEvent event) {
                if (event instanceof NewDataElementReceivedEvent) {
                    NewDataElementReceivedEvent e = (NewDataElementReceivedEvent) event;
                    if (e.getDataElement().getSourceNode().getNodeIdentifier() != networkModuleHelper.getMyNodeIdentifier().getNodeIdentifier()) {
                        networkModuleHelper.debug("New data element registered. Sending beacon.");
                        shouldSendBeacon = true;
                    }
                }
            }
        });

        // On new data element registered in data store
        eventManager.getDataFragmentEventPipe(connectionIdentifier).listen(new SimpleEventListener() {
            public void receiveEvent(SimpleEvent event) {
                if (event instanceof NewDataFragmentsReceived) {
                    NewDataFragmentsReceived e = (NewDataFragmentsReceived) event;
                    for (DataFragment m : e.getNewFragments())
                        if (m.getSourceNode().getNodeIdentifier() != networkModuleHelper.getMyNodeIdentifier().getNodeIdentifier()) {
                            networkModuleHelper.debug("New data fragment registered. Sending beacon.");
                            shouldSendBeacon = true;
                            break;
                        }
                }
            }
        });        
        

        // on new node discovered
        eventManager.getNodeManagerEventPipe().listen(new SimpleEventListener() {
            public void receiveEvent(SimpleEvent event) {
                if (event instanceof NewNodeDiscoveredEvent) {
                    NewNodeDiscoveredEvent e= (NewNodeDiscoveredEvent) event;
                    networkModuleHelper.debug("New node discovered " + e.getIpAddress() + "/" + e.getNodeIdentifier() +". Queuing beacon.");
                    //resetRate(); // Slow start
                    shouldSendBeacon = true;   
                }
            }
        });
        
        // on new subscription
        eventManager.getDefaultPipe().listen(new SimpleEventListener() {
            public void receiveEvent(SimpleEvent event) {
                if (event instanceof NewSubscriptionEvent) {
                    NewSubscriptionEvent e= (NewSubscriptionEvent) event;
                    networkModuleHelper.debug("New subscription added " + e.getTopic().toString() + "/" + e.getSourceConnection() +". Queuing beacon.");
                    shouldSendBeacon = true;
                }
            }
        });


        // On periodic beacon event
        localEventPipe.listen(new SimpleAsyncEventListener() {
            public void receiveEvent(SimpleEvent event) {
                if (event instanceof PeriodicBeaconSendEvent) {
                    // this event should only trigger when there is minimum timerTick ms since we sent something last
                    // since beacons may be sent as a normal reply as well, we have to keep track of them
                    long timeDiff = System.currentTimeMillis() - lastBeaconSent;
                    if (timeDiff < periodicBeaconTime) {
                        try {
                            Thread.sleep(periodicBeaconTime - timeDiff);
                        } catch (InterruptedException ex) {
                            return;
                        }
                        localEventPipe.trigger(event);
                        return;
                    }

                    try {
                        networkModuleHelper.debug(("--- Sending periodic beacon after timer expired --- "));
                        shouldSendBeacon = false;
                        sendBeacon();
                    } catch (MistException ex) {
                        eventManager.throwException(new MistException("Unable to send periodic beacon, retrying in 10 seconds", ex));
                        try {
                            Thread.sleep(10 * 1000);
                        } catch (InterruptedException ex1) {
                            return;
                        }
                        localEventPipe.trigger(event);                        ;
                        return;
                    }
                    try {
                        // trigger new beacon in the future
                        Thread.sleep(periodicBeaconTime + randomMs(timerTick, 0.5));
                    } catch (InterruptedException ex) {
                        return;
                    }
                    localEventPipe.trigger(event);                                           
                }
            }
        });
        localEventPipe.trigger(new PeriodicBeaconSendEvent());
        
        // On periodic neighbor update event
        localEventPipe.listen(new SimpleAsyncEventListener() {
            public void receiveEvent(SimpleEvent event) {
                if (event instanceof PeriodicNeighborUpdateEvent) {
                    try {
                        Thread.sleep(randomMs(timerTick, 0.5));

                        // remove expired nodes
                        Collection<NodeIdentifier> expired = networkModuleHelper.getExpiredNodes(periodicBeaconTime * 3);
                        if (expired != null && !expired.isEmpty()) {
                            eventManager.getNodeManagerEventPipe().trigger(new RemoveNodeEvent(expired, connectionIdentifier));
                        }                    

                        // We send ack from here if we optimize for short messages. Otherwise the ACK
                        // should be delivered from a separate, more frequent, event.

                        sendDataFragments(optimizeForShortMessages);

                    } catch (Exception ex) {
                        eventManager.throwException(new MistException("Unable to send neighbor update. Update aborted.", ex));
                    }
                    localEventPipe.trigger(event); // send neighbor updates every seconds, when available 
                }
            }
        });
        localEventPipe.trigger(new PeriodicNeighborUpdateEvent());
        
        // On periodic ACK. Must run more frequently than neighbor update when the message size in general is larger than the ack size
        if (!optimizeForShortMessages) {
            localEventPipe.listen(new SimpleAsyncEventListener() {
                public void receiveEvent(SimpleEvent event) {
                    if (event instanceof PeriodicAckSendEvent) {
                        int neighbors = networkModuleHelper.getNodes().size() + 1;
                        try {
                            Thread.sleep(randomMs(timerTick / neighbors, 0.5));
                        } catch (InterruptedException ex) {
                            return;
                        }
                        try {
                            if (shouldSendBeacon) {
                                shouldSendBeacon = false;
                                try {
                                    sendBeacon();
                                    networkModuleHelper.debug("Ack beacon sent to " + neighbors + " neighbors");
                                } catch (Exception ex) {
                                    eventManager.throwException(new MistException("Unable to send beacon.", ex));
                                }
                            }
                        } finally {                        
                            localEventPipe.trigger(event); // send neighbor updates every seconds, when available
                        }
                    }
                }
            });
            localEventPipe.trigger(new PeriodicAckSendEvent());            
       }        
    }

    public int getFragmentSize() {
        return fragmentSize;
    }

    public void setConnectionIdentifier(final ConnectionIdentifier connectionIdentifier) {
        this.connectionIdentifier = connectionIdentifier;        
    }

    public ConnectionIdentifier getConnectionIdentifier() {
        return this.connectionIdentifier;
    }

    public void setNodeIdentifier(NodeIdentifier nodeIdentifier) {
        this.nodeIdentifier = nodeIdentifier;
    }

    public NodeIdentifier getNodeIdentifier() {
        return nodeIdentifier;
    }

    public void threadExceptionCallback(Runnable runnable, Exception ex) {
        eventManager.throwException(new MistException("Thread exception from runnable " + runnable.toString(), ex));
    }

    public void threadStoppedCallback(Runnable thread) {
        //emi.triggerEvent(this, new ExceptionEvent(new MistException("Thread stopped: " + thread.toString())));
        MistUtil.debug("Thread stopped: " + thread.toString());
    }


    /**
     *
     *
     * EVENTS
     *
     *
     *
     */

    public class PeriodicBeaconSendEvent implements SimpleEvent {}

    public class PeriodicNeighborUpdateEvent implements SimpleEvent {}    

    public class PeriodicAckSendEvent implements SimpleEvent {}

}
